package com.qf.air.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.qf.air.entity.Air;
import com.qf.air.enums.AirEnum;
import com.qf.air.exception.AirEx;
import com.qf.air.mapper.AirMapper;
import com.qf.air.service.AirService;
import com.qf.air.service.DistrictService;
import com.qf.air.util.R;
import com.qf.air.vo.ResultVO;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpRequest;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.*;

/**
 * @Author: 维多利加
 * @Date: 2021/3/20 11:02
 */
@Service
public class AirServiceImpl implements AirService {

    @Autowired
    private AirMapper airMapper;

    @Autowired
    private RestTemplate restTemplate;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private DistrictService districtService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public PageInfo<Air> findByPageAndDistrictId(Integer page, Integer rows, Integer districtId) {
        PageHelper.startPage(page, rows);
        List<Air> all = airMapper.findByDistrictId(districtId);
        PageInfo<Air> pageInfo = new PageInfo<>(all);
        return pageInfo;
    }

    @Override
    @Transactional
    public void save(Air air) throws JsonProcessingException {
        Integer count = airMapper.save(air);
        if (count != 1) {
            //添加失败，记录日志，抛出异常
            System.out.println("添加信息失败！ air=" + air);
            throw new AirEx(AirEnum.AIR_ADD_ERROR);
        }
        // 数据库添加成功
        // 调用districtService查询对应的区域名称
        String districtName = districtService.findNameById(air.getDistrictId());
        air.setDistrictName(districtName);

        // 发送消息到MQ,让search保存数据到es
        rabbitTemplate.convertAndSend("air_exchange","air.add",objectMapper.writeValueAsString(air));
        /*

        // 调用search模块,将数据保存到ES
        // 封装请求头
        HttpHeaders header = new HttpHeaders();
        header.setContentType(MediaType.APPLICATION_JSON_UTF8);

        // 封装HttpEntity
        HttpEntity httpEntity = new HttpEntity(air,header);

        // 调用restTemplate访问Search模块
        //    http://localhost:9200/search/air/add
        restTemplate.postForObject("http://localhost:9200/search/air/add",httpEntity,String.class);*/

    }

    @Override    // TODO  同步到ES
    @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.DEFAULT, timeout = -1, readOnly = false, rollbackFor = Exception.class, noRollbackFor = ArithmeticException.class)
    public void deleteById(Integer id) {
        int i = airMapper.deleteById(id);

        if (i != 1) {
            System.out.println("删除失败！ id=" + id);
            throw new AirEx(AirEnum.AIR_DELETE_ERROR);
        }

    }

    @Override
    @Transactional    // TODO  同步到ES
    public void updateById(Air air) {
        int count = airMapper.updateById(air);

        if (count != 1) {
            System.out.println("修改失败！ air=" + air);
            throw new AirEx(AirEnum.AIR_UPDATE_ERROR);
        }
    }

    //    http://localhost:9200/search/air/count
    //    http://localhost:9200/search/air/list
    // core!
    @Override
    public ResultVO findByPageAndConditionFromElasticsearch(Integer page, Integer rows, Integer districtId, String monitoringStation, Date startTime, Date endTime) throws IOException {
        //1. 封装查询条件    restTemplate
        Map parameter = new HashMap();    // request body
        parameter.put("page", page);
        parameter.put("rows", rows);
        parameter.put("districtId", districtId);
        parameter.put("monitoringStation", monitoringStation);
        parameter.put("startTime", startTime);
        parameter.put("endTime", endTime);


        //2 封装请求头和请求参数
        HttpHeaders httpHeaders = new HttpHeaders();     // request header
        httpHeaders.setContentType(MediaType.APPLICATION_JSON_UTF8);

        HttpEntity countHttpEntity = new HttpEntity(parameter, httpHeaders);  // request body & request header

        //3. 发送请求到search模块查询数据条数
        Long count = restTemplate.postForObject("http://localhost:9200/search/air/count", countHttpEntity, Long.class);

        //4. 如果数据条数 > 0,继续查询具体的数据
        if (count > 0) {
            HttpEntity listHttpEntity = new HttpEntity(parameter, httpHeaders);  // request body & request header
            String json = restTemplate.postForObject("http://localhost:9200/search/air/list", listHttpEntity, String.class);
            // 由于返回的json字符串在响应给客户端时，有一次对字符串进行了序列化，JSON字符串乱七八糟，页面无法识别
            // {"districtId": 2}   ->  {\"districtId\": 2}  导致出现问题
            List<Map> data = objectMapper.readValue(json, new TypeReference<List<Map>>() {});
            return R.ok(data, count);
        }
        //5. 封装返回
        return R.ok(null, 0L);
    }

    @Override
    public Object findEchartsData() throws IOException {
        //1. SearchRequest
        SearchRequest request = new SearchRequest();

        //2. 设置索引信息
        request.indices("air");

        //3. 设置使用的聚合查询
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        sourceBuilder.aggregation(AggregationBuilders.histogram("xxx").field("pm25").interval(1));

        request.source(sourceBuilder);

        //4. 发送请求获取执行结果
        SearchResponse resp = restHighLevelClient.search(request, RequestOptions.DEFAULT);

        //5. 解析出buckets中的数组
        Histogram xxx = resp.getAggregations().get("xxx");

        List<Map<String, Object>> data = new ArrayList<>();

        for (Histogram.Bucket bucket : xxx.getBuckets()) {
            Map<String, Object> map = new HashMap<>();
            map.put("key", bucket.getKey());
            map.put("docCount", bucket.getDocCount());
            data.add(map);
        }
        return data;
    }
}
